package com.kitten.comment.retry;

import com.kitten.comment.model.dto.PublishCommentMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

/**
 * @author kitten
 */
@Slf4j
@Component
public class SendMqRetryHelper {

    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Resource
    private RetryTemplate retryTemplate;
    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor taskExecutor;

    @Retryable(
            retryFor = { Exception.class },  // 需要重试的异常类型
            maxAttempts = 3,                 // 最大重试次数
            backoff = @Backoff(delay = 1000, multiplier = 2)  // 初始延迟时间 1000ms，每次重试间隔加倍
    )
    public void send(String topic, String body) {
        log.info("==> 开始异步发送 MQ, Topic: {}, publishCommentMqDTO: {}", topic, body);

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(body)
                .build();

        // 同步发送 MQ
        rocketMQTemplate.asyncSend(topic, message, new SendCallback(){
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 发送评论 MQ 成功, Topic: {}, publishCommentMqDTO: {}", topic, message);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 发送评论 MQ 失败, 进入重试: , Topic: {}, publishCommentMqDTO: {}", topic, message, throwable);
                handleRetry(topic, message);
            }
        });
    }

    // Spring Retry 重试消息
    private void handleRetry(String topic, Message<String> message) {
        log.info("==> 重试消息: Topic: {}", topic);
        taskExecutor.submit(() -> {
            try {
                retryTemplate.execute((RetryCallback<Void, RuntimeException>) context -> {
                    log.info("==> 开始重试 MQ 发送, 当前重试次数: {}, 时间: {}", context.getRetryCount() + 1, LocalDateTime.now());
                    // 同步发送 MQ
                    rocketMQTemplate.syncSend(topic, message);
                    return null;
                });
            } catch (Exception e) {
                // 兜底方案
                asyncSendMessageFallback(e, topic, message.getPayload());
            }
        });
    }


    /**
     * 兜底方案: 将发送失败的 MQ 写入数据库，之后，通过定时任务扫表，将发送失败的 MQ 再次发送，最终发送成功后，将该记录物理删除
     */
    @Recover
    public void asyncSendMessageFallback(Exception e, String topic, String body) {
        log.error("==> 多次发送失败, 进入兜底方案, Topic: {}, publishCommentMqDTO: {}", topic, body);

        // TODO:
    }

}
